package com.lagou.edu.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.Map;

/**
 * @author 赵虎
 * @date 2021/4/1 12:11 上午
 */
@Configuration
public class RabbitConfig {

    public final static String ORDER_QUEUE_NAME = "queue.order";
    public final static String ORDER_TTL_QUEUE_NAME = "queue.order.ttl";
    public final static String ORDER_DLX_QUEUE_NAME = "queue.order.dlx";

    public final static String ORDER_EXCHANGE_NAME = "ex.order";
    public final static String ORDER_TTL_EXCHANGE_NAME = "ex.order.ttl";
    public final static String ORDER_DLX_EXCHANGE_NAME = "ex.order.dlx";

    public final static String ORDER_ROUTER_NAME = "router.order";
    public final static String ORDER_TTL_ROUTER_NAME = "router.order.ttl";
    public final static String ORDER_DLX_ROUTER_NAME = "router.order.dlx";


    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public Queue orderQueue(){
        return QueueBuilder.durable(ORDER_QUEUE_NAME).build();
    }

    @Bean
    public Queue ttlQueue(){
        Map<String,Object> map = new HashMap<>();
        map.put("x-message-ttl",10_000);
        map.put("x-dead-letter-exchange", ORDER_DLX_EXCHANGE_NAME);
        map.put("x-dead-letter-routing-key", ORDER_DLX_ROUTER_NAME);
        return new Queue(ORDER_TTL_QUEUE_NAME,true,false,false,map);
    }

    @Bean
    public Queue dlxQueue(){
        return QueueBuilder.durable(ORDER_DLX_QUEUE_NAME).build();
    }

    @Bean
    public Exchange orderExchange(){
        return ExchangeBuilder.directExchange(ORDER_EXCHANGE_NAME).build();
    }

    @Bean
    public Exchange ttlExchange(){
        return ExchangeBuilder.directExchange(ORDER_TTL_EXCHANGE_NAME).build();
    }

    @Bean
    public Exchange dlxExchange(){
        return ExchangeBuilder.directExchange(ORDER_DLX_EXCHANGE_NAME).build();
    }

    @Bean
    public Binding orderBinding(){
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ORDER_ROUTER_NAME).noargs();
    }

    @Bean
    public Binding ttlBinding(){
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with(ORDER_TTL_ROUTER_NAME).noargs();
    }

    @Bean
    public Binding dlxBinding(){
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(ORDER_DLX_ROUTER_NAME).noargs();
    }

    @Bean(name = "rabbitMessageListener")
    public DirectMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory){
        // 写的时候，默认使用DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2个线程
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
        // 设置确认消息的模式
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setPrefetchCount(5);

        // 每个消费者使用一个Channel
        container.setConsumersPerQueue(5);
        //一次确认几个消息
        container.setMessagesPerAck(1);

        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(20);
        //设置该属性，灵活设置并发 ,多线程运行。
        container.setTaskExecutor(taskExecutor);

        return container;
    }

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}
